﻿
#include "dvr_mml_svr.h"
#include "trace_define.h"
#include "transfer_size.h"
#ifdef _WIN32
//#include <winsock2.h>
#else
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <sys/ioctl.h>
//#include "framebuf_arr.h"
#define  INVALID_SOCKET -1
#endif
#include "storage_api.h"
#include "media_frame_hd.h"

using namespace std;
//支持4个用户

//int CStreamerSvr::LIVE_SEND_BIT = 0xF;

CStreamerSvr::CStreamerSvr()
{
	m_rcvBuf = NULL;
	m_sendBuf = NULL;
	if(NULL == m_rcvBuf)
	{
		m_rcvBuf_sz = SIZE_KB(8);
		m_rcvBuf = new unsigned char[m_rcvBuf_sz];
	}

	m_send_sz = NULL;

	m_sock    = 0;
	memset(m_rcvBuf,0,SIZE_KB(4));

	m_live_ch = 0;
	m_bStream_Live = false;
	m_bReplay = false;
	m_bSendSvr = true;
	m_bMix_ch = false;
#ifdef _WIN32
	WORD sockVersion = MAKEWORD(2,2);
	WSADATA wsaData;
	static bool bInit = false;

	if(!bInit)
	{
		if(WSAStartup(sockVersion, &wsaData)!=0)
		{
			return ;
		}
		bInit = true;
	}

#endif
}

CStreamerSvr::~CStreamerSvr()
{

}
int CStreamerSvr::Init(int send_buf_sz,int portNum,bool bSend)
{
	if(NULL == m_rcvBuf)
	{
		m_rcvBuf = new unsigned char[SIZE_KB(4)];
	}

	if(NULL == m_sendBuf)
	{
		m_send_sz = send_buf_sz;
		m_sendBuf  = new char[send_buf_sz];
	}
	memset(m_rcvBuf,0,SIZE_KB(4));
	memset(m_sendBuf,0,m_send_sz);

	//创建套接字
	m_sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
	if(m_sock == INVALID_SOCKET)
	{
		 printf("socket error !");
		 return 0;
	}
	int try_cnt = 0;
	int on = 1;
	while (true)
	{
		int ret = setsockopt(m_sock,SOL_SOCKET,SO_REUSEADDR,(const char*)&on,sizeof(on));
		if(ret < 0)
		{
			 printf("set sockopt fail.!");
		}
		//绑定IP和端口
		sockaddr_in sin;
		sin.sin_family = AF_INET;
		sin.sin_port = htons(portNum);
	#ifdef _WIN32
		sin.sin_addr.S_un.S_addr = INADDR_ANY;
	#else
		sin.sin_addr.s_addr = htonl(INADDR_ANY);
	#endif // _WIN32
		if(bind(m_sock, (struct sockaddr *)&sin, sizeof(sin)) == -1)
		{
			TRACE("bind error %d! m_sock %d try_cnt %d \n",errno,m_sock,try_cnt);
			if(try_cnt >= 3)
			{
				return -1;
			}
			try_cnt++;
			this_thread::sleep_for(chrono::milliseconds(1000));
			continue;
		}
		else
		{
			break;
		}
	};

	int i = 0;
	for(i = 0; i < 4; i++)
	{
		m_stream_client[i].clientID   = 4;
		m_stream_client[i].clientSock = -1;
		m_stream_client[i].pTask      = this;
		m_stream_client[i].stream_ch  = 0;
	}
	m_bSendSvr = bSend;
    TRACE("Init Stream svr Ok m_sock %d .\n",m_sock);
	return 0;
}

int CStreamerSvr::Start()
{
	m_thd_svr = std::thread(&CStreamerSvr::ListenProc, this);
	m_thd_svr.detach();

	return 0;
}

#ifdef _WIN32
int CStreamerSvr::ListenProc()
{
	int max_sock = 0;
	int strLen;
	fd_set  m_fds,copyRead;
	struct timeval tm;
	unsigned long ul=1;
	SOCKADDR_IN clntAddr;
	int clntAddrSz;
	int read_byte = 0;
	int yes = 1;

	m_sockNum = 0;

	max_sock = m_sock;

	if(listen(m_sock,SOCK_SIZE)==-1)
	{
		TRACE_ERR("listen error,err %d ",errno);
		return -1;
	}
	//遍历socks，将所有的元素置于无效的socket
	for(int i=0;i<SOCK_SIZE;i++)
		m_send_sock[i] = INVALID_SOCKET;
	m_sockNum += 1;

	m_send_sock[0]=m_sock;
	FD_ZERO(&m_fds);
	FD_SET(m_sock,&m_fds);
	tm.tv_sec=5000;
	tm.tv_usec=0;
    TRACE_ERR("Strat  MML server accpet client \n ");

	while(1)
	{
		copyRead=m_fds;
		int selResult = select(m_sockNum +1,&copyRead,0,0,&tm);
		if(selResult==-1)
		{
			puts("select error");
		}
		else if(selResult==0)
		{
			//TRACE_ERR("Time out .\n");
			continue;
		}
		else
		{
			//先判断是否是有新的客户端连接
			if(FD_ISSET(m_send_sock[0],&copyRead))
			{
				clntAddrSz = sizeof(clntAddr);
				int clntSock = accept(m_sock,(struct sockaddr*)&clntAddr,&clntAddrSz);
				//将socket设置成非阻塞模式
				ioctlsocket(clntSock,FIONBIO,&ul);

				m_lock.lock();
				for(int i = 0;i<SOCK_SIZE;i++)
				{
					//遍历socks，在元素为无效的socket处插入客户端的socket
					if(m_send_sock[i] == INVALID_SOCKET)
					{
						FD_SET(clntSock,&m_fds);
						fprintf(stderr,"i %d accept new sokcet %d .\n",i,clntSock);
						int ret = StartStreamSvr(clntSock,i);
						if(ret == ERR_OVER_MAXCLIENT)
						{   //超最大用户支持数
#ifdef _WIN32
							closesocket(clntSock);
#else
							close(clntSock);
#endif
						}
						else
						{
							m_send_sock[i] = clntSock;
							m_sockNum++;
						}
						break;
					}
				}
				m_lock.unlock();
			}
			//遍历所有的客户端socket，0的位置为服务端的socket，所以从1开始
			for (int i = 1;i < SOCK_SIZE;i++)
			{
				//如果是无效的socket 不必处理
				if(m_send_sock[i] == INVALID_SOCKET)
					continue;
				if(FD_ISSET(m_send_sock[i],&copyRead))
				{
					strLen=recv(m_send_sock[i],(char *)m_rcvBuf,m_rcvBuf_sz,0);
					if(strLen <= 0)//客户端断开了连接
					{
						closesocket(m_send_sock[i]);
						//从fds删除客户端socket
						FD_CLR(m_send_sock[i],&m_fds);
						//将对应的位置再次置为无效socket
						m_send_sock[i] = INVALID_SOCKET;
						m_sockNum--;
					}
				}
			}
		}
	}
	closesocket(m_sock);
	return 0;
}
#else
int CStreamerSvr::ListenProc()
{
	int max_sock = 0;
	struct timeval tm;
	unsigned long ul=1;
	struct sockaddr_in clntAddr;
	socklen_t clntAddrSz;
	int i = 0;

	m_sockNum = 0;

	max_sock = m_sock;

	if(listen(m_sock,SOCK_SIZE)==-1)
	{
		TRACE_ERR("listen error,m_sock %d err %d ",m_sock,errno);
		return -1;
	}
	//遍历socks，将所有的元素置于无效的socket
	for(int i=0;i<SOCK_SIZE;i++)
		m_send_sock[i] = INVALID_SOCKET;

	m_sockNum = m_sock;
	m_send_sock[0]=m_sock;

	tm.tv_sec=5000;
	tm.tv_usec=0;
	TRACE_ERR("Strat accpet client \n ");
	while(1)
	{
		FD_ZERO(&m_fds);
		FD_SET(m_sock,&m_fds);

		max_sock = m_sock;
		//遍历socket池 找出值最大的socket
		for (i = 0; i < SOCK_SIZE; i++)
		{
			if (m_send_sock[i] != -1)
			{
				//将socket池中的所有socket都添加到事件数组allset中
				FD_SET(m_send_sock[i], &m_fds);
				if (m_send_sock[i] > max_sock)
				{
					max_sock = m_send_sock[i];    //maxfd永远是值最大的socket
				}
			}
		}

		int selResult = select(m_sock +1,&m_fds,0,0,&tm);
		if(selResult < 0)
		{
			printf("select failed ! max_sock %d  error message:%s\n", max_sock,strerror(errno));
			break;
		}
		else if(selResult==0)
		{
		    continue;
		}
		else
		{
			//先判断是否是有新的客户端连接
			if(FD_ISSET(m_send_sock[0],&m_fds))
			{
				clntAddrSz = sizeof(clntAddr);
				int clntSock = accept(m_sock,(struct sockaddr*)&clntAddr,&clntAddrSz);
				//将socket设置成非阻塞模式

				ioctl(clntSock,FIONBIO,&ul);
				for(int i=0;i<SOCK_SIZE;i++)
				{
					//遍历socks，在元素为无效的socket处插入客户端的socket
					if(m_send_sock[i] == INVALID_SOCKET)
					{
						FD_SET(clntSock,&m_fds);
						m_send_sock[i]=clntSock;
						m_sockNum++;
						StartStreamSvr(clntSock,i);
						break;
					}
				}
			}

			//遍历所有的客户端socket，0的位置为服务端的socket，所以从1开始
			for (int i=1;i < SOCK_SIZE;i++)
			{
				//如果是无效的socket 不必处理
				if(m_send_sock[i]==INVALID_SOCKET)
					continue;
//				if(FD_ISSET(m_send_sock[i],&copyRead))
//				{
//					strLen=recv(m_send_sock[i],(char *)m_rcvBuf,m_rcvBuf_sz,0);
//					if(strLen <= 0)//客户端断开了连接
//					{
//
//						close(m_send_sock[i]);
//						//从fds删除客户端socket
//						FD_CLR(m_send_sock[i],&fds);
//						//将对应的位置再次置为无效socket
//						m_send_sock[i] == INVALID_SOCKET;
//						m_sockNum--;
//					}
//				}
			}
		}
	}

	close(m_sock);
	return 0;
}
#endif
int CStreamerSvr::StartStreamSvr(int clntSock,int index)
{
	int ret = 0;

	STREAM_SVR_INFO_S * pSvrInfo = GetFreeClientInfo(clntSock,index);
	if(pSvrInfo == NULL)
	{   //超最大用户支持数
		TRACE_ERR("No Free clientID.\n");
		return -1;
	}
	m_stop_stream = false;
	m_thd = std::thread(&CStreamerSvr::StreamProc, pSvrInfo);
	m_thd.detach();

	return ret;
}


int CStreamerSvr::RcvData(char * buf ,int buf_sz,int conn_fd)
{
	fd_set      mFdr;
	timeval  mTimeOut;

	FD_ZERO( &mFdr );
	FD_SET( conn_fd, &mFdr );
	int ret = 0;
	mTimeOut.tv_sec = 1;
	mTimeOut.tv_usec = 100000;
	int flags = select( conn_fd + 1, &mFdr, NULL, NULL, &mTimeOut );
	if( flags > 0 )
	{
		ret = recv(conn_fd, buf, buf_sz, 0);
		if(ret == 0 )
		{
			TRACE_ERR("Can not recive data. clientSock %d errno %d  \n",conn_fd, errno);
			if(errno != 0)
			{
               return -1;
			}
			return 0;
		}

	}
	return ret;
}

int CStreamerSvr::SendData(unsigned char* buf, int len, int flags,int conn_fd)
{
	int sendBytes = 0;
	fd_set  mFdw;
	FD_ZERO( &mFdw );
	FD_SET( conn_fd, &mFdw );
	timeval mTimeOut;
	mTimeOut.tv_sec = 0;
	mTimeOut.tv_usec = 5*1000;
	select( conn_fd + 1, NULL, &mFdw, NULL, &mTimeOut );
	if ( !FD_ISSET(conn_fd, &mFdw) )
	{
		return SOCKETNODATA;
	}

	sendBytes = send(conn_fd, (char*)buf, len, flags);

	return sendBytes;
}

int  CStreamerSvr::ReadFrameData(unsigned char * pFrm,int buf_sz,int ch, int clientID)
{
	int ret = 0;
	if(!m_bReplay)
	{
		ret = STFS_FifoReadFrame(pFrm, buf_sz, ch, clientID);// m_frameStrategy->PlaybackGetFrame(pFrm, buf_sz, ch, 0);
	}
	else
	{
		ret = STFS_ReplayFifoReadFrame(pFrm,buf_sz,ch,0);
	}

	return ret;
}


STREAM_SVR_INFO_S * CStreamerSvr::GetFreeClientInfo(int sock,int index)
{
	int i = 0;

	for(i = 0; i < 4; i++)
	{
		if(-1 == m_stream_client[i].clientSock)
		{
			m_stream_client[i].clientSock = sock;
			m_stream_client[i].clientID  = 0;//i + 1; //起始ID设计为1，5为RTSP
			m_stream_client[i].pTask     = (void*)this;
			m_stream_client[i].stream_ch = 0;
			m_stream_client[i].sock_idx = index;
			m_stream_client[i].bMix_ch = false;
			m_stream_client[i].send_ch_bit = 0x01; //默认4通道
			m_stream_client[i].last_ch = 0;
			m_stream_client[i].replay_id = 0; //回放ID
			m_stream_client[i].isReplay =1; 
			printf("GetFreeClientInfo clientID %d .\n", m_stream_client[i].clientID);
			return &m_stream_client[i];
		}
	}

	return NULL;
}



void CStreamerSvr::DestroyClient(int sock)
{
	int i = 0;

	for(i = 0; i < 4; i++)
	{
		if(sock == m_stream_client[i].clientSock)
		{
			m_stream_client[i].clientSock = -1;
			m_stream_client[i].replay_id = 0;
            TRACE("DestroyClient %d i %d .\n",sock,i);
			break;
		}
	}

}

int GetNexCh(int & startCh,int chBit)
{
	int dest_ch = 0;
	int i = 0;
	bool bFound  = false;
	if(startCh == 31)
	{
		startCh = 0;
	}

	for(i = startCh + 1 ; i < 16 ; i++)
	{
		if((chBit>>i) & 0x1)
		{
			dest_ch = i;
			bFound = true;
			break;
		}
	}

	if(!bFound)
	{
		for(i = 0; i < 16 ; i++)
		{
			if((chBit>>i) & 0x1)
			{
				dest_ch = i;
				bFound = true;
				break;
			}
		}
	}

	startCh = dest_ch;
    return dest_ch;
}

int CStreamerSvr:: StreamProc(void * pParam)
{
	STREAM_SVR_INFO_S* pClientSock = (STREAM_SVR_INFO_S * )pParam;
	CStreamerSvr * pTask = (CStreamerSvr*)pClientSock->pTask;
#ifdef _WIN32
	SOCKET conn_fd;
#else
	int conn_fd;
#endif

	conn_fd = pClientSock->clientSock;
	
	int    ret  = 0;
	int    frmLen = 0;
	int    buf_sz = pTask->m_send_sz;
	int    data_pos  =0;
	int    err = 0;
	struct timeval tmpT;
	int64_t curTm;
	unsigned char * pFrm = pClientSock->frmBuf;
	pTask->LIVE_SEND_BIT = 0x03;
	
	pTask->m_bStream_Live = true;
	int dest_ch = 0;
	int clientID = pClientSock->clientID;

	fprintf(stderr,"Start stream proc thread.isReplay %d m_bStream_Live %d \n",pClientSock->isReplay,pTask->m_bStream_Live);
	int try_cnt = 10;
	int read_cnt = 0;
	char file_path[128] = {0};

    char time_str[64] = {0};

	int64_t dvr_tm =0 ;

	GetDvrCurTime(&dvr_tm);

	int file_sz = 0;
	pFrm = new unsigned char[SIZE_KB(500)];
	RD_FRAME_H * pHead = (RD_FRAME_H*)pFrm;
	int send_buffer_size = 256 * 1024; // 设置发送缓冲区大小为1MB
	if (setsockopt(conn_fd, SOL_SOCKET, SO_SNDBUF, &send_buffer_size, sizeof(send_buffer_size)) == -1) {
		perror("setsockopt");
	//	exit(1);
	}
	while (pTask->m_bStream_Live)
	{
		//中间视频流传输
		if(pTask->m_stop_stream)
		{
			TRACE("stop stream send .\n");
			break;
		}
		if(pClientSock->isReplay)
		{
			dest_ch = 0;//GetNexCh(pClientSock->last_ch,0xF);
		}
		else
		{
			clientID = pClientSock->clientID;
			dest_ch = GetNexCh(pClientSock->last_ch,pTask->LIVE_SEND_BIT);
		}

		
		if (!pClientSock->isReplay)
		{
			frmLen = ret = STFS_FifoReadFrame_live(pFrm, buf_sz, dest_ch, clientID);// m_frameStrategy->PlaybackGetFrame(pFrm, buf_sz, ch, 0);
		}
		else
		{
			frmLen = ret = STFS_ReplayFifoReadFrame(pFrm, buf_sz, dest_ch, 0);
		}

		if(ret <= 0)
		{
			this_thread::sleep_for(chrono::milliseconds(10));
		 TRACE("ReadFrameData fail dest_ch %d ret %d clientID %d m_bReplay %d .\n", dest_ch,ret, clientID,pTask->m_bReplay);
			continue;
		}
		
		read_cnt++;
		RD_FRAME_H* pHead = (RD_FRAME_H*)pFrm;
		int64_t pts = pHead->pts;
		if(pHead->frameType== AFrame )
		{
			//TRACE("ch %02d clientID %d  FNO %d m_bReplay %d AFrame pts %lld .\n",dest_ch, clientID, frmTool.GetFrameNo(),pTask->m_bReplay,pts);
			continue;
		}

		 
		if (pClientSock->isReplay)
		{  //局域网内使用比较合适
			//pts = STFS_PlaybackCurTm(dest_ch, 0);
			bool noWait = (pClientSock->send_cnt++ < 10) ? 1:0;
			STFS_ReplaySyncWait(pts, 0, dest_ch, 0,noWait);
			if (pHead->frameType == IFrame )
			{
				PrintTimeMs(pts, 0, "IFrame");
			//	ShowHexData((char*)pFrm,96,32,1,"Iframe ");
				printf("ch %02d clientID %d  FNO %d Send Frame len %d curTm %lld  buf_sz %d bWait %d send_cnt %d  \n",
					dest_ch, clientID,pHead->frameNo,pHead->dataLen, pHead->pts,buf_sz,noWait,pClientSock->send_cnt);
			}
		}
		else
		{
			gettimeofday(&tmpT, NULL);//时间戳
			curTm = (int64_t)tmpT.tv_sec * 1000 + (int64_t)tmpT.tv_usec/1000;
			//int64_t cur_pts = frmTool.GetFramePts();
			int64_t diff = curTm - pHead->pts;
			//int64_t diff = pts;
			if(pHead->frameType== IFrame)//||
				//diff > 90 )
			{
				printf("ch %02d clientID %d  FNO %d Send Frame len %d diff %lld curTm %lld pts %lld ",
					dest_ch, clientID,pHead->frameNo,
					pHead->dataLen,diff,curTm, pHead->pts);
		    	PrintTimeMs(pts, 0, "IFrame ");
			}	
		}

		// if(frmLen > buf_sz)
		// {
		// 	continue;
		// }

		data_pos = 0;
		int send_len = 0;
		int i = 0;
		int leaf_len = 0;
		int send_sz = 64<<10;

		while (data_pos < frmLen)
		{
			int try_cnt = 150;
			bool bSendOk = false;
			 
            leaf_len = frmLen - data_pos;
			if(leaf_len > send_sz )
			{
				leaf_len = send_sz;
			}

			for(i = 0; i < try_cnt; i++)
			{
				ret =  send(conn_fd, (char*)pFrm + data_pos, leaf_len, 0); //pTask->SendData(pFrm + data_pos,leaf_len,0,conn_fd);

				if(ret >= leaf_len)
				{
					bSendOk = true;
					break;
				}
				if(ret <= 0)
				{
					err = errno;
					if(err == 32)
					{
						goto SEND_DONE;
					}
					if(err == 104)
					{
						goto SEND_DONE;
					}
					TRACE("Send frame  last error %d conn_fd %d ret %d send_len %d  frmLen %d leaf_len %d try_cnt %d .\n",
						   err,conn_fd,ret,send_len,frmLen,leaf_len,i);
					this_thread::sleep_for(chrono::milliseconds(10));
					if(i == try_cnt - 1)
					{
						goto SEND_DONE;
					}
				}
			}
			if(!bSendOk)
			{
				break;
			}
			send_len += ret;
			if(send_len  >= frmLen)
			{
				break;
			}
			data_pos += ret;
		}

	}

SEND_DONE:
	if(pFrm != NULL)
	{
		delete [] pFrm;
	}
	pTask->m_bStream_Live = false;
#ifdef _WIN32
	closesocket(conn_fd);
#else
	close(conn_fd);
	int idx = pTask->m_sockNum;
	//从fds删除客户端socket
//	FD_CLR(pTask->m_send_sock[idx],&pTask->m_fds);
	//将对应的位置再次置为无效socket
	pTask->m_send_sock[idx -1 ] = INVALID_SOCKET;
	pTask->m_sockNum--;

#endif


	pTask->DestroyClient(conn_fd);
	TRACE("CStreamerSvr:: StreamProc Exit stream thread.\n");

	return 0;
}

int CStreamerSvr:: StreamHeartProc(void * pParam)
{
	return 0;
}

int CStreamerSvr::StreamProcRecv(void * pParam)
{
	STREAM_SVR_INFO_S* pClientSock = (STREAM_SVR_INFO_S * )pParam;
	CStreamerSvr * pTask = (CStreamerSvr*)pClientSock->pTask;
#ifdef _WIN32
	SOCKET conn_fd;
#else
	int conn_fd;
#endif

	conn_fd = pClientSock->clientSock;
	unsigned char * pFrm = pClientSock->frmBuf;
	int    ret  = 0;
	int    pktLen = 0;
	int    buf_sz = pTask->m_send_sz;
	int    try_cnt = 3;

//	LHN_FRAME_H * pHead = (LHN_FRAME_H*)pFrm;

	pTask->m_bStream_Live = true;


	//CMultiCHData* mulData = CMultiCHData::getSingleton();
	while (pTask->m_bStream_Live)
	{

		pktLen = ret = pTask->RcvData((char*)pFrm,buf_sz,conn_fd);

		if(ret <0 )
		{
			if( 0 == try_cnt--)
			{
				goto SEND_DONE;
			}
			TRACE("Rcv ret %d err %d .\n",ret,errno);
 
			this_thread::sleep_for(chrono::milliseconds(10));
		}
		if(ret > 0)
		{
			//ret = mulData->FIFO_WRITE_FRAME(pFrm,pktLen,0,0,0,true);
			TRACE("Rcv data size %d .\n",pktLen);
		}
		this_thread::sleep_for(chrono::milliseconds(30));
	}

SEND_DONE:
	TRACE("SEND_DONE stream thread.\n");
	pTask->m_lock.lock();
	int idx = pClientSock->sock_idx;
	if(pTask->m_send_sock[idx] !=INVALID_SOCKET )
	{
#ifdef _WIN32
		closesocket(conn_fd);
#else
		close(conn_fd);
#endif

		//从fds删除客户端socket
		//	FD_CLR(pTask->m_send_sock[idx],&pTask->m_fds);
		//将对应的位置再次置为无效socket
		pTask->m_send_sock[idx] = INVALID_SOCKET;
		pTask->m_sockNum--;

		pTask->DestroyClient(conn_fd);


	}
	pTask->m_lock.unlock();
	TRACE("CStreamerSvr:: StreamProcRecv Exit stream thread.\n");
	return 0;
}
